package com.qyer.search.engine;

import static com.qyer.commons.utils.CommonUtils.LEFT;
import static com.qyer.commons.utils.CommonUtils.RIGHT;
import static com.qyer.commons.utils.CommonUtils.splitList;

import com.qyer.search.cache.RedisCacheHolder;
import com.qyer.search.model.SearchParameter;
import com.qyer.search.searchplan.SQPlan;
import com.qyer.words.model.Word;
import org.apache.commons.collections4.CollectionUtils;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.RecursiveTask;

/**
 * User: Z J Wu Date: 2016/01/21 Time: 11:36 Package: com.qyer.search.engine
 */
public class RecursiveSearchTask extends RecursiveTask<List<String>> {

  private static final int TIMEOUT_IN_SECONDS = 60;

  private AbstractESEngine searchEngine;

  private List<SQPlan> plans;

  private boolean showHighlight;
  private boolean simpleResult;

  protected String getCacheKey(SearchParameter sp, final List<Word> words) {
    int result = sp.getConditionCacheKey();
    if (CollectionUtils.isEmpty(words)) {
      return String.valueOf(result);
    }
    List<Word> newList = new ArrayList<>(words);
    Collections.sort(newList);
    result = 31 * result + newList.hashCode();
    return "sr." + (simpleResult ? "0" : "1") + String.valueOf(result);
  }

  public RecursiveSearchTask(AbstractESEngine searchEngine, List<SQPlan> plans,
                             boolean showHighlight, boolean simpleResult) {
    if (CollectionUtils.isEmpty(plans)) {
      throw new RuntimeException("Empty plan.");
    }
    this.searchEngine = searchEngine;
    this.plans = plans;
    this.showHighlight = showHighlight;
    this.simpleResult = simpleResult;
  }
  
  private List<String> atomicPlanSearch(SQPlan plan) {
    SearchParameter sp = plan.getSearchParameter();
    String key = getCacheKey(sp, plan.getWords());

    List<String> result = RedisCacheHolder.getInstance().getCache(key);
    if (CollectionUtils.isEmpty(result)) {

      result = searchEngine.executePlanExtractAsString(sp, plan, showHighlight, simpleResult);
      RedisCacheHolder.getInstance().putCache(key, result, TIMEOUT_IN_SECONDS);
    } else {
      AbstractESEngine.logDebugMap(sp, true, plan);
    }
    if (result == null) {
      result = Collections.emptyList();
    }
    return result;
  }

  @Override protected List<String> compute() {
    List<String> result;
    if (CollectionUtils.isEmpty(plans)) {
      result = Collections.emptyList();
    } else if (plans.size() == 1) {
      result = atomicPlanSearch(plans.get(0));
    } else {
      List<SQPlan> l = splitList(plans, LEFT);
      RecursiveSearchTask left = new RecursiveSearchTask(searchEngine, l, showHighlight,
                                                         simpleResult);
      l = splitList(plans, RIGHT);
      RecursiveSearchTask right = new RecursiveSearchTask(searchEngine, l, showHighlight,
                                                          simpleResult);
      left.fork();
      right.fork();
      int mergeSize = 0;
      List<String> leftResult = left.join(), rightResult = right.join();
      mergeSize += CollectionUtils.isEmpty(leftResult) ? 0 : leftResult.size();
      mergeSize += CollectionUtils.isEmpty(rightResult) ? 0 : rightResult.size();
      result = new ArrayList(mergeSize);
      result.addAll(leftResult);
      result.addAll(rightResult);
    }
    return result;
  }

}
